package com.sqllog.admin.sql.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.sqllog.admin.cache.CanalClientCache;
import com.sqllog.admin.framework.tools.SequenceUtil;
import com.sqllog.admin.sql.constants.SqlLogConstants;
import com.sqllog.admin.sql.datasource.LogSaveDataSourceOption;
import com.sqllog.admin.sql.datasource.MongoDbSaveDataSourceOption;
import com.sqllog.admin.sql.domain.SqlLogInfo;
import com.sqllog.admin.sql.entity.LogDatabaseManager;
import com.sqllog.admin.sql.service.CanalClientConfigInfoService;
import com.sqllog.admin.sql.service.CanalClientConsumerLogService;
import com.sqllog.admin.sql.service.LogDatabaseManagerService;
import com.sqllog.admin.utils.CommonUtils;
import com.sqllog.admin.utils.SpringContextUtils;
import com.sqllog.entity.RowChangeInfo;
import com.sqllog.listener.SimpleCanalEventListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * 基础数据监听
 *
 * @author chennengcheng
 * @date 2022-01-12 11:14
 * @remark
 */

@Slf4j
public class DataBaseLogListener extends SimpleCanalEventListener {

    private LogSaveDataSourceOption logSaveDataSourceOption = SpringContextUtils.getBean(MongoDbSaveDataSourceOption.class);

    private LogDatabaseManagerService logDatabaseManagerService = SpringContextUtils.getBean(LogDatabaseManagerService.class);

    private CanalClientConsumerLogService canalClientConsumerLogService = SpringContextUtils.getBean(CanalClientConsumerLogService.class);

    private CanalClientConfigInfoService canalClientConfigInfoService = SpringContextUtils.getBean(CanalClientConfigInfoService.class);

    // 每千次 记录一条日志
    private long LAST_SAVE_TIME = 0;

    @Override
    public void processSqlMessage(RowChangeInfo rowChangeInfo) {
        String host = rowChangeInfo.getHost();
        String destination = rowChangeInfo.getDestination();
        String schemaName = rowChangeInfo.getSchemaName();
        String tableName = rowChangeInfo.getTableName();
        CanalEntry.RowChange rowChange = rowChangeInfo.getRowChange();
        doOption(host, destination, schemaName, tableName, rowChange);
    }
    /**
     * 分表保存
     *
     * @param dataBaseName
     * @param tableName
     * @param sqlListeners
     */
    public void saveByTableName(String destination, String dataBaseName, String tableName, List<SqlLogInfo> sqlListeners) {
        checkAndCreateCollection(destination, dataBaseName, tableName);
        logSaveDataSourceOption.insert(dataBaseName, tableName, CommonUtils.getCurrentMonthName(), sqlListeners);
    }

    /**
     * 校验集合是否存在如果存在则跳过，如果不存在则创建
     *
     * @param dataBaseName
     * @param tableName
     */
    public void checkAndCreateCollection(String destination, String dataBaseName, String tableName) {
        String month = CommonUtils.getCurrentMonthName();
        String collectionName = CommonUtils.generateTableNameByYearOfMonth(dataBaseName, tableName, month);
        boolean collectionExist = CanalClientCache.logDatabaseManageMap.containsKey(collectionName);
        // 更新注解中 集合名
        if (!collectionExist) {
            logSaveDataSourceOption.createTable(dataBaseName, tableName, month);
            LogDatabaseManager logDatabaseManager = logDatabaseManagerService.insert(destination, dataBaseName, tableName, month);
            CanalClientCache.logDatabaseManageMap.put(logDatabaseManager.getQueryName(), Collections.singletonList(logDatabaseManager));
            logSaveDataSourceOption.createTableIndex(dataBaseName, tableName, month, "rowIdValue");
        }
    }

    public void doOption(String host, String destination, String schemaName, String tableName, CanalEntry.RowChange rowChange) {
        List<SqlLogInfo> sqlListenerEntities = new ArrayList<>();
        String logStr = "";
        if (CanalEntry.EventType.INSERT.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在新增 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = this.insertOption(rowChange);
        }

        if (CanalEntry.EventType.DELETE.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在删除 schemaName: " + schemaName + "tableName: " +tableName;
            sqlListenerEntities = this.deleteOption(rowChange);
        }

        if (CanalEntry.EventType.UPDATE.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在修改 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = this.updateOption(rowChange);
        }

        if (CanalEntry.EventType.ALTER.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在修改表结构 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = this.alertOption(rowChange);
        }

        if (CanalEntry.EventType.CREATE.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在创建表 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = this.createTableOption(rowChange);
        }

        if (CanalEntry.EventType.CINDEX.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在创建索引 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = createIndexOption(rowChange);
        }

        if (CanalEntry.EventType.DINDEX.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在删除索引 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = dropIndexOption(rowChange);
        }

        if (CanalEntry.EventType.ERASE.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在清空表 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = dropTableOption(rowChange);
        }

        if (CanalEntry.EventType.RENAME.equals(rowChange.getEventType())) {
            logStr = logStr + "canal 正在重命名表 schemaName: " + schemaName + "tableName: " + tableName;
            sqlListenerEntities = renameOption(rowChange);
        }

        log.debug(logStr);
        saveCanalClientConsumerLog(host, destination);
        saveByTableName(destination, schemaName, tableName, sqlListenerEntities);
    }

    private void saveCanalClientConsumerLog(String host, String destination){
        if (this.LAST_SAVE_TIME == 0) {
            this.LAST_SAVE_TIME = System.currentTimeMillis();
            return;
        }
        long currentTime = System.currentTimeMillis();
        long differTime = currentTime - this.LAST_SAVE_TIME;
        this.LAST_SAVE_TIME = currentTime;
        // 如果大于 30 秒 就记录一次
//        if (differTime > 30) {
//            CanalClientConfigInfo canalClientConfigInfo = canalClientConfigInfoService.selectByHostAndInstanceName(host, destination);
//            String id = canalClientConfigInfo.getId();
//            canalClientConsumerLogService.saveCanalClientConsumerLog(id, "正在消费中 host: " + host + " destination: " + destination);
//        }
    }

    @Override
    public void callBackForException(RowChangeInfo rowChangeInfo, Exception e) {
        log.error(JSON.toJSONString(e));
    }
    /**
     * 删除操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> deleteOption(CanalEntry.RowChange rowChange) {
        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
        List<SqlLogInfo> sqlRecordEntitys = new ArrayList<>();
        for (CanalEntry.RowData rowData : rowDataList) {
            if (CollectionUtils.isEmpty(rowData.getBeforeColumnsList())) {
                continue;
            }
            Map<String, String> updateMap = new HashMap<>(16);
            StringBuilder idKey = new StringBuilder();
            StringBuilder idValue = new StringBuilder();
            for (CanalEntry.Column c : rowData.getBeforeColumnsList()) {
                if (c.getIsKey()) {
                    idKey.append(c.getName());
                    idValue.append(c.getValue());
                    updateMap.put(c.getName(), c.getValue());
                    break;
                }
            }

            SqlLogInfo sqlRecordEntity =
                    CommonUtils.getTableSqlListenerEntity(
                            SequenceUtil.makeStringId(),
                            SqlLogConstants.SqlType.DELETE,
                            SqlLogConstants.ObjType.ROW, updateMap, idKey.toString(), idValue.toString());
            sqlRecordEntitys.add(sqlRecordEntity);
        }

        return sqlRecordEntitys;
    }

    /**
     * 新增操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> insertOption(CanalEntry.RowChange rowChange) {
        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
        List<SqlLogInfo> sqlRecordEntitys = new ArrayList<>();
        for (CanalEntry.RowData rowData : rowDataList) {
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            Map<String, String> updateMap = new HashMap<>(16);
            String rowIdKey = StrUtil.EMPTY;
            String rowIdValue = StrUtil.EMPTY;
            for (CanalEntry.Column column : afterColumnsList) {
                if (column.getIsKey()) {
                    rowIdKey = column.getName();
                    rowIdValue = column.getValue();
                }

                updateMap.put(column.getName(), column.getValue());
            }

            SqlLogInfo sqlRecordEntity =
                    CommonUtils.getTableSqlListenerEntity(
                            SequenceUtil.makeStringId(),
                            SqlLogConstants.SqlType.INSERT,
                            SqlLogConstants.ObjType.ROW, updateMap, rowIdKey, rowIdValue);
            sqlRecordEntitys.add(sqlRecordEntity);
        }

        return sqlRecordEntitys;
    }

    /**
     * 更新操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> updateOption(CanalEntry.RowChange rowChange) {
        List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
        List<SqlLogInfo> sqlRecordEntitys = new ArrayList<>();
        for (CanalEntry.RowData rowData : rowDataList) {
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            // 更新map 值
            Map<String, String> updateMap = new HashMap<>(16);
            String rowIdKey = StrUtil.EMPTY;
            String rowIdValue = StrUtil.EMPTY;
            for (CanalEntry.Column column : afterColumnsList) {
                // 是否是条件
                if (column.getIsKey()) {
                    rowIdKey = column.getName();
                    rowIdValue = column.getValue();
                } else {
                    updateMap.put(column.getName(), column.getValue());
                }
            }

            SqlLogInfo sqlRecordEntity =
                    CommonUtils.getTableSqlListenerEntity(
                            SequenceUtil.makeStringId(),
                            SqlLogConstants.SqlType.UPDATE,
                            SqlLogConstants.ObjType.ROW, updateMap, rowIdKey, rowIdValue);

            sqlRecordEntitys.add(sqlRecordEntity);
        }

        return sqlRecordEntitys;
    }

    /**
     * 修改表操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> alertOption(CanalEntry.RowChange rowChange) {
        Map<String, String> valueMap = new HashMap<>();
        valueMap.put("sql", rowChange.getSql());
        SqlLogInfo sqlRecordEntity =
                CommonUtils.getTableSqlListenerEntity(
                        SequenceUtil.makeStringId(),
                        SqlLogConstants.SqlType.ALERT,
                        SqlLogConstants.ObjType.TABLE, valueMap);
        return CollectionUtil.newArrayList(sqlRecordEntity);
    }

    /**
     * 创建表操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> createTableOption(CanalEntry.RowChange rowChange) {
        Map<String, String> valueMap = new HashMap<>();
        valueMap.put("sql", rowChange.getSql());
        SqlLogInfo sqlRecordEntity =
                CommonUtils.getTableSqlListenerEntity(
                        SequenceUtil.makeStringId(),
                        SqlLogConstants.SqlType.CREATE_TABLE,
                        SqlLogConstants.ObjType.TABLE, valueMap);
        return CollectionUtil.newArrayList(sqlRecordEntity);
    }

    /**
     * 创建索引操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> createIndexOption( CanalEntry.RowChange rowChange) {
        Map<String, String> valueMap = new HashMap<>();
        valueMap.put("sql", rowChange.getSql());
        SqlLogInfo sqlRecordEntity =
                CommonUtils.getTableSqlListenerEntity(
                        SequenceUtil.makeStringId(),
                        SqlLogConstants.SqlType.CREATE_INDEX,
                        SqlLogConstants.ObjType.TABLE, valueMap);
        return CollectionUtil.newArrayList(sqlRecordEntity);
    }

    /**
     * 删除索引操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> dropIndexOption(CanalEntry.RowChange rowChange) {
        Map<String, String> valueMap = new HashMap<>();
        valueMap.put("sql", rowChange.getSql());
        SqlLogInfo sqlRecordEntity =
                CommonUtils.getTableSqlListenerEntity(
                        SequenceUtil.makeStringId(),
                        SqlLogConstants.SqlType.DROP_INDEX,
                        SqlLogConstants.ObjType.TABLE, valueMap);
        return CollectionUtil.newArrayList(sqlRecordEntity);
    }

    /**
     * 清空表操作
     *
     * @param rowChange
     * @return
     */
    public List<SqlLogInfo> dropTableOption(CanalEntry.RowChange rowChange) {
        Map<String, String> valueMap = new HashMap<>();
        valueMap.put("sql", rowChange.getSql());
        SqlLogInfo sqlRecordEntity =
                CommonUtils.getTableSqlListenerEntity(
                        SequenceUtil.makeStringId(),
                        SqlLogConstants.SqlType.DROP_TABLE,
                        SqlLogConstants.ObjType.TABLE, valueMap);
        return CollectionUtil.newArrayList(sqlRecordEntity);
    }

    /**
     * 重命名表操作

     * @return
     */
    public List<SqlLogInfo> renameOption(CanalEntry.RowChange rowChange) {
        Map<String, String> valueMap = new HashMap<>();
        valueMap.put("sql", rowChange.getSql());
        SqlLogInfo sqlRecordEntity =
                CommonUtils.getTableSqlListenerEntity(
                        SequenceUtil.makeStringId(),
                        SqlLogConstants.SqlType.RENAME,
                        SqlLogConstants.ObjType.TABLE, valueMap);
        return CollectionUtil.newArrayList(sqlRecordEntity);
    }
}
